package cn.doitedu.features;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_广告点击率预估特征数据加工 {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中的 行为日志: dwd-events
        tenv.executeSql(
                "create table dwd_events_kafka (                              \n" +
                        "     user_id           string                        \n" +
                        "    ,event_id           string                       \n" +
                        "    ,action_time        bigint                       \n" +
                        "    ,properties         map<string,string>           \n" +
                        "    ,ad_tracking_id as properties['ad_tracking_id']  \n" +
                        "    ,rt as to_timestamp_ltz(action_time,3)           \n" +
                        "    ,watermark for rt as rt                          \n" +
                        ") WITH (                                             \n" +
                        "    'connector' = 'kafka',                          \n" +
                        "    'topic' = 'dwd-events',                         \n" +
                        "    'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                        "    'properties.group.id' = 'doit44_g1',            \n" +
                        "    'scan.startup.mode' = 'latest-offset',          \n" +
                        "    'value.format' = 'json',                        \n" +
                        "    'value.fields-include' = 'EXCEPT_KEY'           \n" +
                        ")                                                    ");

        // 建表，映射hbase中的广告请求日志表： ad_request_log
        tenv.executeSql(
                " create table ad_request_log_hbase(                  \n "+
                        "   ad_tracking_id  string,                   \n "+
                        "   f ROW<                                    \n "+
                        " 	       requestTime  bigint,               \n "+
                        " 	       platform string,                   \n "+
                        " 	       pageId string,                     \n "+
                        " 	       locationType string,               \n "+
                        " 	       mediaType string,                  \n "+
                        " 	       adids string,               \n "+
                        " 	       ad_features string,         \n "+
                        " 	       user_features string,       \n "+
                        " 	       device_features string,     \n "+
                        " 	       poster_features string      \n "+
                        " 	>,                                        \n "+
                        " 	PRIMARY KEY(ad_tracking_id) not enforced           \n "+
                        " ) WITH (                                             \n "+
                        "  'connector' = 'hbase-2.2',                          \n "+
                        "  'table-name' = 'ad_request_log',                    \n "+
                        "  'zookeeper.quorum' = 'doitedu:2181'                 \n "+
                        " )                                                    \n ");


        // cep 匹配
        tenv.executeSql(
                        " create temporary view ad_cep AS                             "+
                        " with tmp as (                                               "+
                        " select * from dwd_events_kafka                              "+
                        " where event_id in ('ad_show','ad_click','ad_transform')     "+
                        " )                                                           "+
                        " SELECT                                                      "+
                        "     *, proctime() as pt                                     "+
                        " from tmp                                                    "+
                        " match_recognize(                                            "+
                        "     PARTITION BY ad_tracking_id                             "+
                        " 	ORDER BY rt                                               "+
                        "     MEASURES                                                "+
                        "       A.user_id as uid,                                     "+
                        "       A.ad_tracking_id as tk_id,                            "+
                        " 		A.action_time as show_time,                           "+
                        " 		B.action_time as click_time,                          "+
                        " 		C.action_time as transform_time                       "+
                        "     ONE ROW PER MATCH                                       "+
                        "     AFTER MATCH SKIP PAST LAST ROW                          "+
                        "     PATTERN(A B C)                                           "+
                        "     DEFINE                                                  "+
                        " 	    A AS A.event_id='ad_show',                            "+
                        " 	    B AS B.event_id='ad_click',                           "+
                        " 	    C AS C.event_id='ad_transform'                        "+
                        " )                                                           "
        );


        // cep匹配的结果关联 广告请求日志
        tenv.executeSql(
                        " SELECT                                                           "+
                        "   a.uid,                                                         "+
                        " 	a.tk_id,                                                       "+
                        " 	a.show_time,                                                   "+
                        " 	a.click_time,                                                  "+
                        " 	a.transform_time,                                              "+
                        " 	b.f.requestTime,                                               "+
                        " 	b.f.platform,                                                  "+
                        " 	b.f.pageId,                                                    "+
                        " 	b.f.locationType,                                              "+
                        " 	b.f.mediaType,                                                 "+
                        " 	b.f.adids,                                                     "+
                        " 	b.f.ad_features,                                               "+
                        " 	b.f.user_features,                                             "+
                        " 	b.f.device_features,                                           "+
                        " 	b.f.poster_features                                            "+
                        " FROM  ad_cep a                                                   "+
                        " LEFT JOIN ad_request_log_hbase FOR SYSTEM_TIME AS OF a.pt AS b   "+
                        " ON a.tk_id = b.ad_tracking_id                                    "
        ).print();
    }
}
